package com.godyao.mall.seckill.service.impl;

import cn.hutool.core.date.DateUtil;
import cn.hutool.core.util.ObjectUtil;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.godyao.mall.core.utils.OauthUtils;
import com.godyao.mall.core.model.LoginVal;
import com.godyao.mall.core.model.ResultCode;
import com.godyao.mall.core.model.ResultMsg;
import com.godyao.mall.rabbitmq.service.ISendMsgService;
import com.godyao.mall.seckill.model.SeckillActivity;
import com.godyao.mall.seckill.model.msg.SeckillMessage;
import com.godyao.mall.seckill.model.response.SeckillActivityResponse;
import com.godyao.mall.seckill.mq.config.RabbitMqConfig;
import com.godyao.mall.seckill.repository.SeckillRepository;
import com.godyao.mall.seckill.service.ISeckillService;
import lombok.RequiredArgsConstructor;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.data.domain.Page;
import org.springframework.data.domain.Pageable;
import org.springframework.data.jpa.domain.Specification;
import org.springframework.data.redis.core.HashOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ValueOperations;
import org.springframework.data.redis.core.script.RedisScript;
import org.springframework.stereotype.Service;

import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

import static com.godyao.mall.core.constants.RedisConstant.*;

/**
 * @author godyao
 * @date 2022/4/1
 */
@Slf4j
@Service
@RequiredArgsConstructor
public class SeckillServiceImpl implements ISeckillService, InitializingBean {
    private final SeckillRepository seckillRepository;

    private final RedisTemplate redisTemplate;

    private final RedissonClient redissonClient;

    private final ISendMsgService sendMsgService;

    private final RedisScript buyScript;
    /**
     * 内存标记 减少对redis的访问
     */
    private static final Map<Long, Boolean> GOODS_CAN_BY_MAP = new HashMap<>();

    @Override
    public Page<SeckillActivityResponse> findFrontAll(Specification specification, Pageable pageable) {
        return seckillRepository.findAll(specification, pageable);
    }

    @Override
    public String generateSecKillCode(Long goodsId) {
        final LoginVal currentUser = OauthUtils.getCurrentUser();
        String key = SEC_KILL_URL_CODE_PREFIX + currentUser.getUserId() + ":" + goodsId;
        final String code = UUID.randomUUID().toString();
        // 设置code有效期
        redisTemplate.opsForValue().set(key, code, 10, TimeUnit.MINUTES);
        return code;
    }

    @SneakyThrows
    @Override
    public ResultMsg buy(Long goodsId) {
        if (GOODS_CAN_BY_MAP.getOrDefault(goodsId, true)) {
            return ResultMsg.result(ResultCode.NO_STOCK);
        }
        final ValueOperations valueOperations = redisTemplate.opsForValue();
        final String userId = OauthUtils.getCurrentUser().getUserId();
        String userGoodKey = userId + ":" + goodsId;
        if (valueOperations.getOperations().hasKey(SEC_KILL_GOODS_KEY_PREFIX + userGoodKey)) {
            return ResultMsg.result(ResultCode.SEC_KILL_LIMIT);
        }

        final String key = SEC_KILL_KEY_PREFIX + goodsId;
        final HashOperations hashOperations = redisTemplate.opsForHash();
        if (!hashOperations.getOperations().hasKey(key) && checkGoodsValid(goodsId)) {
            // 获取不到则验证商品信息从数据库加载 判断商品是不是秒杀商品以及是否在秒杀时间范围内
            final RLock lock = redissonClient.getLock(SEC_KILL_LOCK);
            try {
                final boolean b = lock.tryLock(10, TimeUnit.SECONDS);
                if (b) {
                    // 获取到锁 从数据库load数据到缓存
                    final SeckillActivity seckillActivity = seckillRepository.findByGoodsId(goodsId).get();
                    if (ObjectUtil.isNotNull(seckillActivity)) {
                        final Map<String, Object> goodsInfo = new HashMap<>();
                        goodsInfo.put("stock", seckillActivity.getStock());
                        goodsInfo.put("start_time", seckillActivity.getStartTime());
                        goodsInfo.put("end_time", seckillActivity.getEndTime());
                        hashOperations
                                .putAll(key, goodsInfo);
                    } else {
                        // 对于不合法的商品数据 防止缓存穿透（数据库和缓存都没有） 设置过期时间
                        // 原子性对业务影响不大，如果有影响可以使用lua脚本
                        hashOperations
                                .putAll(key, new HashMap());
                        redisTemplate.expire(key, 5, TimeUnit.SECONDS);
                    }
                }
            } catch (InterruptedException e) {
                log.error("秒杀失败，msg={}", e.getMessage());
                return ResultMsg.resultFail();
            } finally {
                if (null != lock && lock.isHeldByCurrentThread()) lock.unlock();
            }
        }

        // 这里增加用户超买判断
        // 1.判断商品是否有效(在活动期间的商品)
        // 2.redis setKey 商品购买用户集合判断用户多买
        // 3.判断库存数是否足够
        final String buyedUserKey = SEC_KILL_KEY_PREFIX + "buyed:" + goodsId;
        if (checkGoodsValid(hashOperations, key)) {
            final Long res = (Long) redisTemplate
                    .execute(buyScript, Arrays.asList(key, buyedUserKey), userId);
            if (res == -2) {
                return ResultMsg.result(ResultCode.SEC_KILL_LIMIT);
            }
            if (res > 0) {
                // 下单操作放到rabbitmq中 削峰 解耦
                final SeckillMessage seckillMessage = new SeckillMessage(userId, goodsId);
                final String data = new ObjectMapper().writeValueAsString(seckillMessage);
                sendMsgService.sendMsg(RabbitMqConfig.EXCHANGE_NAME, RabbitMqConfig.ROUTE_KEY, data, null);
                return ResultMsg.result(ResultCode.WAITING);
            }
            return ResultMsg.result(ResultCode.NO_STOCK);

        } else {
            GOODS_CAN_BY_MAP.put(goodsId, true);
            hashOperations.increment(key, "stock", 1);
            return ResultMsg.result(ResultCode.NO_STOCK);
        }
    }


    @Override
    public boolean checkUrlCode(String code, Long goodsId) {
        final LoginVal currentUser = OauthUtils.getCurrentUser();
        String key = SEC_KILL_URL_CODE_PREFIX + currentUser.getUserId() + ":" + goodsId;
        return code.equals((String) redisTemplate.opsForValue().get(key));
    }

    @Override
    public void del(Long id) {
        deleteRedisGoodsInfoById(id);
        seckillRepository.deleteById(id);
    }

    @Override
    public SeckillActivity save(SeckillActivity seckillActivity) {
        final SeckillActivity save = seckillRepository.save(seckillActivity);
        // 同步redis
        updateRedisGoodsInfo(save);
        return save;
    }

    /**
     * 缓存预热
     * @throws Exception
     */
    @Override
    public void afterPropertiesSet() {
        seckillRepository.findAll().parallelStream().peek(seckillActivity -> {
            updateRedisGoodsInfo(seckillActivity);
        }).collect(Collectors.toList());
    }

    private void updateRedisGoodsInfo(SeckillActivity seckillActivity) {
        final Map<String, Object> goodsInfo = new HashMap<>();
        goodsInfo.put("stock", seckillActivity.getStock());
        goodsInfo.put("start_time", seckillActivity.getStartTime());
        goodsInfo.put("end_time", seckillActivity.getEndTime());

        redisTemplate.opsForHash()
                .putAll(SEC_KILL_KEY_PREFIX+ seckillActivity.getGoodsId(), goodsInfo);
        // false即库存不为空
        if (seckillActivity.getStock() > 0) {
            // 标识该商品已经售罄
            GOODS_CAN_BY_MAP.put(seckillActivity.getGoodsId(), false);
        } else {
            GOODS_CAN_BY_MAP.put(seckillActivity.getGoodsId(), true);
        }
    }

    private void deleteRedisGoodsInfoById(Long goodsId) {
        redisTemplate.delete(SEC_KILL_KEY_PREFIX + goodsId);
        GOODS_CAN_BY_MAP.remove(goodsId);
    }


    /**
     * 读取缓存数据判断商品有效性 减少数据库压力
     * @param hashOperations
     * @param key
     * @return
     */
    public boolean checkGoodsValid(HashOperations hashOperations, String key) {
        // 从缓存拿到商品信息 验证商品有效性
        if (hashOperations.getOperations().hasKey(key)) {
            final Integer stock = (Integer) hashOperations.get(key, "stock");
            final Date startTime = (Date) hashOperations.get(key, "start_time");
            final Date endTime = (Date) hashOperations.get(key, "end_time");
            if (DateUtil.compare(new Date(), startTime) > 0 && DateUtil.compare(new Date(), endTime) < 0 && stock > 0) {
                return true;
            }
        }
        return false;
    }


    /**
     * 缓存加载条件
     * @param goodsId
     * @return
     */
    private boolean checkGoodsValid(Long goodsId) {
        return seckillRepository.findGoodsInvalid(goodsId) > 0;
    }
}
